一：RAG 的工作原理

阶段一：准备阶段（建立知识库）

1. 数据接入：收集各种文档（PDF、Word、网页等）
2. 文档解析：提取文本内容
3. 文档分割：将长文档切分成小片段
4. 向量化：将文本转换为数学向量
5. 存储：将向量存入专门的数据库

阶段二：问答阶段（智能应答）

1. 用户提问：输入问题
2. 问题向量化：将问题也转换成向量
3. 相似度检索：在向量数据库中寻找最相关的文档片段
4. 构建增强提示：将检索到的文档 + 原始问题组合成新的提示
5. 生成答案：大语言模型基于增强后的提示生成最终答案

二：RAG 完整执行流程详解

2.1. 知识库构建阶段（离线）

1. 数据源接入

目标：从各种来源收集原始数据

常见数据源：

• 本地文件（PDF、Word、TXT、Markdown）
• 数据库记录
• 网页内容
• API 接口数据
• 企业内部文档

data_sources = {
    "本地文件": [".pdf", ".docx", ".txt"],
    "网页内容": ["博客", "文档", "新闻"],
    "数据库": ["MySQL", "MongoDB"],
    "API 接口": ["企业知识库", "云存储"]
}

2. 文档解析

目标：从原始文件中提取纯文本内容

处理类型：

• PDF → 提取文字、表格
• HTML → 去除标签，保留正文
• Word → 解析文档结构
• 图片 → OCR 文字识别

技术工具：

• PyPDF2、pdfplumber（PDF 解析）
• BeautifulSoup（HTML 解析）
• python-docx（Word 解析）
• Tesseract（OCR）

# 使用 LangChain 进行 Word 文档解析示例
from langchain_community.document_loaders import Docx2txtLoader

# 初始化加载器，传入 Word 文档路径，加载文档为 Document 对象
loader = Docx2txtLoader("公司假期制度.docx")
documents = loader.load()

# 查看加载结果
print(f"加载了 {len(documents)} 个文档片段")
print(f"第一段内容预览: {documents[0].page_content[:200]}...")
print(f"元数据: {documents[0].metadata}")

3. 文档分割

目标：将长文档切分成适合处理的知识片段（chunks）

为什么需要分割？

• 大语言模型有输入长度限制
• 提高检索精度
• 避免信息过载

常用分割方法：

• 固定长度分割
• 按段落 / 句子分割
• 语义分割（按主题变化）

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 创建文本分割器，按照固定长度切分，并保留重叠部分
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,      # 每个文本块的最大字符数
    chunk_overlap=100,   # 相邻文本块之间的重叠字符数（保持上下文连贯性）
    separators=["\n\n", "\n", "。", "！", "？", "，", ""],  # 定义了分割符的优先级顺序
    add_start_index=True,  # 记录每个块在原文档中的起始位置
)

# 执行分割
all_splits = text_splitter.split_documents(documents)
print(f"分割后得到 {len(all_splits)} 个文本块")

4. 词向量化

• 1. 将文本转成一组向量：每个下标 i，代表一个维度
• 2. 整个数组对应一个 n 维空间的一个点，即文本向量又叫 Embeddings
• 3. 向量之间可以计算距离，距离远近对应语义相似度大小

目标：将文本转换为数值向量（数学表示）

向量化原理：

文本："RAG 技术很强大"
↓ 向量化模型（Embedding 模型）
向量：[0.12, -0.45, 0.78, ..., 0.33] (384 维向量)

常用模型：

• OpenAI text-embedding-3
• 国产模型：M3E、BGE 等

使用 OpenAI 模型进行 Embedding 向量化

from dotenv import load_dotenv
import os

load_dotenv()

OPENAI_API_KEY: str = os.getenv('OPENAI_API_KEY')
OPENAI_BASE_URL: str = os.getenv('OPENAI_BASE_URL')

# 默认模型配置
model_name = "text-embedding-3-small"

from openai import OpenAI
client = OpenAI(base_url=OPENAI_BASE_URL, api_key=OPENAI_API_KEY)

# 准备需要向量化的文本
text_to_embed = "今天天气真好，我们出去散步吧！"

# 调用 Embedding 模型
response = client.embeddings.create(
    model=model_name,  # 使用默认模型配置中的模型名称
    input=text_to_embed # 需要向量化的文本
)

# 从响应中提取生成的向量
embedding_vector = response.data[0].embedding
print(f"生成向量的维度：{len(embedding_vector)}")
print(f"向量预览：{embedding_vector[:5]}...")

5. 文档存储

目标：将向量和原文存储到专用数据库中

向量数据库选择：

• ChromaDB（轻量级，易用） www.trychroma.com/
• Pinecone（云服务，高性能） www.pinecone.io/
• Faiss（高并发，gpu 加速）https://github.com/facebookresearch/faiss
• Milvus（毫秒级搜索万亿级向量数据）http://milvus.io/

from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings

# 初始化嵌入模型
embeddings = HuggingFaceEmbeddings(
    model_name='BAAI/bge-large-zh-v1.5',
    model_kwargs={'device': 'cpu'}  # 使用 GPU 可改为 'cuda'
)

# 创建向量数据库
vectorstore = FAISS.from_documents(all_splits, embeddings)

# 保存到本地（可选）
vectorstore.save_local("word_doc_faiss_index")

2.2. 问答推理阶段（在线）

1. 用户提问

用户输入自然语言问题："RAG 技术的主要优势有哪些？"

2. 文档检索

过程：

1. 将用户提出的问题向量化
2. 在向量数据库中搜索相似度最高的前 N 个文档
3. 返回最相关的文档片段

检索算法：

• 余弦相似度
• 欧氏距离
• 近似最近邻搜索

向量间的相似度计算

from openai import OpenAI
import os
import numpy as np
# NumPy 库中的线性代数模块
import numpy.linalg as norm
from dotenv import load_dotenv
load_dotenv()

client = OpenAI()

def cos_sim(a,b):
    """计算余弦相似度，数值越大越相似"""
    return np.dot(a, b) / (norm.norm(a) * norm.norm(b))

def l2_dist(a,b):
    """计算欧氏距离，数值越小越相似"""
    return norm.norm(np.asarray(a) - np.asarray(b))

def get_embedding(texts, model="text-embedding-3-small"):
    """
    获取文本的 embedding 向量

    参数:
        texts: 字符串或字符串列表
        model: 模型名称

    返回:
        如果 texts 是单个字符串，返回单个向量（列表）
        如果 texts 是字符串列表，返回向量列表（列表的列表）
    """
    # 确保 texts 是列表格式
    if isinstance(texts, str):
        texts = [texts]

    response = client.embeddings.create(
        model=model,
        input=texts
    )

    # 返回所有文档的向量
    embeddings = [item.embedding for item in response.data]

    # 如果只有一个文本，返回单个向量；否则返回向量列表
    return embeddings[0] if len(embeddings) == 1 else embeddings

# 支持跨语言计算
query = "Artificial Intelligence in Education"
# query = "人工智能在教育中的应用"
documents = [
    "机器学习算法正在改变传统教学模式，实现个性化学习路径推荐",
    "深度学习技术在自然语言处理领域的突破，使得智能辅导系统更加精准",
    "计算机视觉技术帮助在线教育平台实现学生专注度监测和行为分析",
    "大数据分析为教育决策提供数据支持，优化课程设计和教学资源配置",
    "虚拟现实和增强现实技术创造沉浸式学习体验，提高学习效果和参与度",
]

# 进行转换 Embedding 向量
# query 是单个字符串，get_embedding 返回单个向量
query_vec = get_embedding(query)

# documents 是列表，get_embedding 返回向量列表
doc_vecs = get_embedding(documents)

print(f"查询向量维度: {len(query_vec)}")
print(f"文档数量: {len(doc_vecs)}")
print(f"每个文档向量维度: {len(doc_vecs[0])}")

向量数据库搜索查询文档

from langchain_community.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings

# 1. 加载已保存的 FAISS 向量数据库
embeddings = HuggingFaceEmbeddings(
    model_name='BAAI/bge-large-zh-v1.5',
    model_kwargs={'device': 'cpu'}
)

vectorstore = FAISS.load_local(
    "word_doc_faiss_index",  # 你之前保存的目录名
    embeddings,
    allow_dangerous_deserialization=True  # 必要的安全警告确认机制，明确告诉用户"反序列化可能执行任意代码"
    # LangChain 团队为了保护用户，默认禁止加载可能不安全的序列化数据，除非用户明确确认风险
)

# 2. 执行相似度搜索
query = "调休的申请流程是什么？"

# docs = vectorstore.similarity_search(query, k=3)  # 返回最相关的 3 个文档
# print(f"查询: '{query}'")
# print(f"找到 {len(docs)} 个相关文档:\n")
#
# for i, doc in enumerate(docs):
#     print(f"--- 结果 {i + 1} ---")
#     print(f"内容: {doc.page_content}")
#     print(f"来源: {doc.metadata.get('source', 'Unknown')}")
#     print()

docs_with_scores = vectorstore.similarity_search_with_score(query, k=3)  # 带相似度分数的搜索

# 3. 打印检索结果
for doc, score in docs_with_scores:
    print(f"相似度: {score:.4f} - 内容: {doc.page_content}...")
    print("==================================================")

3. 提示增强

目标：将检索到的文档与用户问题组合成增强提示

提示模板示例：

基于以下上下文信息，请回答问题：

【相关文档1】
RAG技术可以接入最新数据，减少模型幻觉...

【相关文档2】
RAG系统能够提供可验证的信息来源...

【相关文档3】
通过检索增强，模型可以回答专业领域问题...

问题：调休的申请流程是什么？
请根据上述上下文回答：

def build_rag_prompt(query, retrieved_docs):
    """
    构建RAG提示词
    """
    # 1. 拼接检索到的文档内容
    context = "\n\n".join([doc.page_content for doc in retrieved_docs])

    # 2. 构建提示词模板
    prompt = f"""请根据以下上下文信息回答问题。如果上下文中有答案，请基于上下文回答；如果没有，请说明。
        上下文信息：
        {context}
        问题：{query}
        请给出详细、准确的回答：
    """
    return prompt

4. 答案生成

目标：大语言模型基于增强提示生成最终答案

生成过程：增强提示 → 大语言模型 → 生成答案

模型选择：

• GPT 系列
• 文心一言、通义千问
• Llama、ChatGLM 等开源模型

def call_llm_with_rag(query, vectorstore, llm_model):
    """
    完整的 RAG 流程：检索 + 提示词构建 + 调用大模型
    """
    # 1. 检索相关文档
    retrieved_docs = vectorstore.similarity_search(query, k=3)
    # 2. 构建提示词
    prompt = build_rag_prompt(query, retrieved_docs)
    # 3. 调用大模型
    response = llm_model.invoke(prompt)
    return response.content

from langchain_openai import ChatOpenAI

# 初始化 OpenAI 模型
llm_model = ChatOpenAI(
    model="gpt-3.5-turbo",
    openai_api_key=OPENAI_API_KEY,  # 替换为你的 API 密钥
    temperature=0.7
)
# 使用示例
query = "调休的申请流程是什么？"
result = call_llm_with_rag(query, vectorstore, llm_model)
print("用户提问：" + query)
print("模型最终回答：" + result)

2.3. 完整流程演示

def build_rag_prompt(query, retrieved_docs):
    """
    构建RAG提示词
    """
    # 1. 拼接检索到的文档内容
    context = "\n\n".join([doc.page_content for doc in retrieved_docs])
    # 2. 构建提示词模板
    prompt = f"""请根据以下上下文信息回答问题。如果上下文中有答案，请基于上下文回答；如果没有，请说明。
        上下文信息：
        {context}
        问题：{query}
        请给出详细、准确的回答：
    """
    return prompt

# 使用 LangChain 进行 Word 文档解析示例
from langchain_community.document_loaders import Docx2txtLoader
# 文档分割
from langchain.text_splitter import RecursiveCharacterTextSplitter
# 向量数据库
from langchain_community.vectorstores import FAISS
# 向量化 Embedding
from langchain_huggingface import HuggingFaceEmbeddings

# 初始化加载器，传入 Word 文档路径，加载文档为 Document 对象
loader = Docx2txtLoader("公司假期制度.docx")
documents = loader.load()

# 创建文本分割器，按照固定长度切分，并保留重叠部分
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,      # 每个文本块的最大字符数
    chunk_overlap=100,   # 相邻文本块之间的重叠字符数（保持上下文连贯性）
    separators=["\n\n", "\n", "。", "！", "？", "，", ""],  # 定义了分割符的优先级顺序
    add_start_index=True,  # 记录每个块在原文档中的起始位置
)

# 执行分割
all_splits = text_splitter.split_documents(documents)

# 初始化 Embedding 模型
embeddings = HuggingFaceEmbeddings(
    model_name='BAAI/bge-large-zh-v1.5',
    model_kwargs={'device': 'cpu'}  # 使用 GPU 可改为 'cuda'
)

# 创建向量数据库
vectorstore = FAISS.from_documents(all_splits, embeddings)

# 保存到本地（可选）
vectorstore.save_local("word_doc_faiss_index")

# 加载本地向量数据库存储
vectorstore = FAISS.load_local(
    "word_doc_faiss_index",  # 你之前保存的目录名
    embeddings,
    allow_dangerous_deserialization=True  # 必要的安全警告确认机制，明确告诉用户"反序列化可能执行任意代码"
    # LangChain 团队为了保护用户，默认禁止加载可能不安全的序列化数据，除非用户明确确认风险
)

# 用户提问
query = "调休的申请流程是什么？"

# 检索相关文档
retrieved_docs = vectorstore.similarity_search(query, k=3)

# 构建提示词
prompt = build_rag_prompt(query, retrieved_docs)

# 调用大模型回答内容
response = llm_model.invoke(prompt)

print("用户提问：" + query)
print("模型最终回答：" + response.content)

特别注意：

1. 大模型个数和类别：整个 RAG 流程涉及到两类大模型
   • 用于转换成向量的 Embedding 模型（用户提问阶段和知识库入库阶段的模型必须保持一致）
   • 用于生成答案的 LLM 模型
2. 语言模型应用场景：LLM 模型只负责最终将检索到的知识和用户问题进行整合后，回复用户
   • 知识库的数据搭建工作不涉及 LLM 模型
3. RAG 最终目的：这一系列操作都是为了整合好合适的 prompt 提示词，最终交给 LLM
   • prompt 是与 LLM 大模型唯一的交互方式

三：RAG 具体实践落地策略

1. 手动搭建 RAG 引擎
2. 低代码平台搭建 （Coze、Dify、N8N）
3. 使用 GLM、OpenAI Responses API 等进行快速实现
4. 使用 LangChain、LlamaIndex 等开源项目快速搭建

四：实际部署考虑

在选择 RAG 应用场景时，需要考虑：

1. 数据质量：是否有高质量、结构化的知识源？
2. 更新频率：知识需要多频繁更新？
3. 准确度要求：错误的代价有多大？
4. 技术复杂度：是否有相应的技术能力？

五：RAG 应用的优化方案

想要真正进一步优化 RAG 的效果，需要考虑：

1. 优化文本分割：尝试不同的文本分割器（如按句子、递归字符等）和块大小、重叠长度
2. 重排序（Re-ranking）：使用重排序模型对检索到的文档进行重新排序，以提高顶部文档的相关性
3. 多轮对话与历史管理：在对话中维护历史消息，并将历史信息纳入当前查询的上下文中
4. 混合检索：结合关键词检索（如 BM25）和向量检索，以及其他数据库，以提高检索的召回率
5. 评估与迭代: 构建评估集，对 RAG 系统进行定量评估，从而指导优化方向